package com.we.risk.smsInfo.atoV0;

import com.rrd.hive.udf.sms.CvSmsParserUDF;
import com.rrd.hive.udf.CaesarEncodeUDF;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.utils.ParamCheck;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import static com.we.risk.smsInfo.atoV0.AtoSmsV0.*;


public class AtoSmsRichFlatMapFunc extends RichFlatMapFunction<String, String> {
     ValueState<Integer> smsListSizeVState;
     SimpleDateFormat sdf;
//     Logger LOG ;

    @Override
    public void open(Configuration parameters) throws Exception {
        smsListSizeVState =
                getRuntimeContext().getState(new ValueStateDescriptor<Integer>("smsListSize", Integer.class));
        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    }


    @Override
    public void flatMap(String jason, Collector<String> out) throws Exception {
        try {
            Long flink_proc_time= System.currentTimeMillis();
            JSONArray resJsonArray = new JSONArray();
            Integer tmpSmsSize = smsListSizeVState.value();
            int lastSmsSize = (tmpSmsSize == null ? 0 : tmpSmsSize);

            JSONObject dataJson = JSONObject.parseObject(JSONObject.parseObject(jason).getString("dataJson"));

            JSONObject commonJson = new JSONObject();
            commonJson.put(OUTKEYBY, dataJson.getString(INPUTKEYBY)); //获取userKey
            commonJson.put(OUTJOBID, JSONObject.parseObject(dataJson.getString("job")).getString(INPUTJOBID)); //获取jobID
            commonJson.put(RECVTIME, dataJson.getLong(INPUTTIME));//获取timestamp

            JSONObject data = JSONObject.parseObject(dataJson.getString("data"));
            JSONArray smsList = data.getJSONArray(SMSLIST);
            ParamCheck.isNull(smsList, SMSLIST);
            int size = smsList.size();
            CvSmsParserUDF cvSmsParserUDF = new CvSmsParserUDF();

            for (int i = 0; i < size; i++) {
                JSONObject smsJSON = smsList.getJSONObject(i);


                String sendType = smsJSON.getString("type");
                String body = smsJSON.getString("body");
                Long sendDate = Long.parseLong(smsJSON.getString("date"));
                String address = smsJSON.getString("address");


                Map<String, String> udfResMap = cvSmsParserUDF.evaluate(body, address, sendType);

                commonJson.put("flink_proc_time",flink_proc_time);
                commonJson.put("send_type", sendType);
                commonJson.put("send_date", sdf.format(new Date(sendDate)));
                commonJson.put("if_read",smsJSON.getString("read"));
                commonJson.put("opposite_num",new CaesarEncodeUDF().evaluate(address));
                commonJson.put("fin_type", udfResMap.get("fin_type"));
                commonJson.put("src_sms_flags",udfResMap.get("src_sms_flags") );
                commonJson.put("sub_type",udfResMap.get("sub_type") );
                commonJson.put("org_name",udfResMap.get("org_name"));
                commonJson.put("org_loan",udfResMap.get("org_loan"));
                commonJson.put("body",body);
                commonJson.put("trim_body_flags",udfResMap.get("trim_body_flags") );
                commonJson.put("org_name_flags",udfResMap.get("org_name_flags") );
                commonJson.put("trade_type",udfResMap.get("trade_type") );
                commonJson.put("org_type",udfResMap.get("org_type")  );
                commonJson.put("sms_tag",udfResMap.get("sms_tag")  );
                commonJson.put("phone_tag",udfResMap.get("phone_tag"));
                commonJson.put("sms_bank_tag",udfResMap.get("sms_bank_tag"));
                commonJson.put("salary_income",udfResMap.get("salary_income")  );
                commonJson.put("row_num", i);
                commonJson.put("is_active", 1);
                System.out.println("AtoSms输出-----\n"+commonJson.toJSONString());

                resJsonArray.add(commonJson);
            }

            if (size < lastSmsSize) {
                for (int i = size; i < lastSmsSize; i++) {
                    JSONObject expiredCall = new JSONObject();
                    expiredCall.putAll(commonJson);
                    Long label_output_time = System.currentTimeMillis();
                    expiredCall.put("label_output_time",label_output_time);
                    expiredCall.put("row_num", i);
                    expiredCall.put("is_active", 0);

                    resJsonArray.add(expiredCall);
                }

            }
            smsListSizeVState.update(size);
            out.collect(resJsonArray.toString());

        } catch (Exception e) {
                LOG.error("AtoSmsRichFlatMapFunc Exception: " + e.toString());
                LOG.error("AtoSmsRichFlatMapFunc data: " + jason);
            e.printStackTrace(System.out);
        }
    }

}

